package ApplicationStartTest.KafkaConnect.config;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.Optional;

@Slf4j
public class Listener {

    @KafkaListener(topics = {"${spring.kafka.template.default-topic}"}) //listen  default topic
    public void listen(ConsumerRecord<String, String> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            try {
                String messageWrapper = JSON.parseObject(message.toString(),String.class);
                System.out.println("listens: " + messageWrapper);
            } catch (Exception e) {
                log.error("将接收到的消息保存到数据库时异常, 消息:{}, 异常:{}",message.toString(),e);
            }
        }
    }
}
